//package com.bs.config;
//
//import com.bs.deserializer.DataDeserialization;
//import com.bs.sink.RedisSink;
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.springframework.context.annotation.Configuration;
//
//import javax.annotation.PostConstruct;
//
///**
// * @author bingshao
// * @date 2022/11/3
// **/
//@Configuration
//public class CdcStreamMysqlToRedisConfig {
//
//
//    @PostConstruct
//    public void initMysqlToRedisFlink() throws Exception {
////        org.apache.flink.configuration.Configuration config = new org.apache.flink.configuration.Configuration();
////        config.setString("execution.savepoint.path", "C:\\checkpoint\\7aee97e318b5e75a50cd1012cb49644a\\chk-1");
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//                .hostname("110.42.233.197")
//                .port(3306)
//                .databaseList("bingshao")
//                .tableList("bingshao.user")
//                .username("root")
//                .password("bs1354361838")
//                // 自定义反序列化方式
//                .deserializer(new DataDeserialization())
//                .startupOptions(StartupOptions.initial())
//                .build();
//        env.enableCheckpointing(60 * 1000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000L);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 2 * 1000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint");
//        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "bingshao source")
//                .setParallelism(4)
////                .addSink(new RedisSink()).setParallelism(1);
//                .print().setParallelism(1);
//        env.execute("Stream-mysql-to-redis");
//    }
//}
